package io.gitee.bowen_lee.lbwio.group;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * 多路复用器线程<br/>
 * 一个线程对应一个selector <br/>
 * 在多线程下.该程序把并发客户端分配到不同selector里面执行<br/>
 * 即 : 一个客户端有且仅有一个 selector 与之对应.
 * 
 * @author Bowen
 * @since 2022-01-05 10:59:28
 */
public class SelectorThread implements Runnable {

  Selector selector = null;

  Thread curentThread;

  Channel channel = null;

  SelectorThreadGroup stg;

  public SelectorThread(SelectorThreadGroup stg) {
    try {
      this.selector = Selector.open();
      this.stg = stg;
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  @Override
  public void run() {
    curentThread = Thread.currentThread();
    // 1. select
    try {
      while (true) {
        // 方法是阻塞的 需要使用 wakeup() 处理
        System.out.println(Thread.currentThread().getId() + "select前 " + selector.keys().size());
        int selcetNum = selector.select();
        System.out.println(Thread.currentThread().getId() + "select后 " + selector.keys().size());

        // 2. selectKeys
        if (selcetNum > 0) {
          Set<SelectionKey> keys = selector.selectedKeys();
          Iterator<SelectionKey> iterator = keys.iterator();
          while (iterator.hasNext()) {
            SelectionKey key = iterator.next();
            // 迭代器中删除当前key 防止重复处理
            iterator.remove();
            // 注册操作
            if (key.isAcceptable()) {
              acceptHandle(key);
            }
            // R 操作
            else if (key.isReadable()) {
              readHandle(key);
            }
            // W 操作
            else if (key.isWritable()) {
              writeHandle(key);
            }
          }
        }

        // 3. run all task
        if (channel != null) {
          if (channel instanceof ServerSocketChannel) {
            ServerSocketChannel ssc = (ServerSocketChannel) channel;
            ssc.register(selector, SelectionKey.OP_ACCEPT);
          } else if (channel instanceof SocketChannel) {
            SocketChannel sc = (SocketChannel) channel;
            // 注册一个 栈外4096大小的 读操作到 红黑树中
            sc.register(selector, SelectionKey.OP_READ, ByteBuffer.allocateDirect(4096));
          }
          channel = null;
        }
      }
    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  private void writeHandle(SelectionKey key) {
  }

  private void readHandle(SelectionKey key) {
    ByteBuffer buf = (ByteBuffer) key.attachment();
    SocketChannel client = (SocketChannel) key.channel();
    buf.clear();// 复位 不是删除数据.而是把 指针置于 0 位置
    while (true) {
      try {
        int num = client.read(buf);
        if (num > 0) {
          buf.flip();
          for (int i = 0; i < buf.limit(); i++) {
            System.out.print((char) buf.get(i));
          }
          System.out.println("");
          while (buf.hasRemaining()) {
            client.write(buf);
          }
          buf.clear();
        } else if (num == 0) {
          break;
        } else {
          // 客户端断开连接
          System.out.println("客户端断开了" + client.getRemoteAddress());
          key.cancel();
          client.close();
          break;
        }

      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }

  /**
   * 处理注册/接收客户端
   * 
   * @param key
   */
  private void acceptHandle(SelectionKey key) {
    ServerSocketChannel server = (ServerSocketChannel) key.channel();

    try {
      System.out.println(Thread.currentThread().getId() + "执行accpet() ");
      SocketChannel client = server.accept(); // 执行accept() 操作
      client.configureBlocking(false); // 设置非阻塞
      // 准备注册
      stg.serverRegister(client);

    } catch (IOException e) {
      e.printStackTrace();
    }
  }

}
